package com.xiguaji.tiktok.task.blogger;

import com.gugee.tiktok.common.model.dal.blogger.BloggerMainRepository;
import com.gugee.tiktok.common.model.mysql.BloggerLog;
import com.gugee.tiktok.data.common.msg.BloggerMessage;
import com.xiguaji.tiktok.task.common.config.AppConfig;
import com.xiguaji.tiktok.task.common.job.TaskParam;
import com.xiguaji.tiktok.task.common.producer.AbstractProducerHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * Created with IntelliJ IDEA.
 * Description:星图播主的采集更新
 * User: luolifeng
 * Date: 2020-11-09 19:28
 */
@Component
public class MarkerBloggerService extends AbstractProducerHandler<BloggerMessage>{
    @Autowired
    BloggerMainRepository bloggerMainRepository;
    @Autowired
    @Qualifier("kafkaTemplateBlogger")
    private KafkaTemplate<Long, BloggerMessage> kafkaTemplateBlogger;
    @Autowired
    AppConfig appConfig;
    @Override
    public List<BloggerMessage> findMessageObject(BloggerLog bloggerLog, TaskParam taskParam) {
        return bloggerMainRepository.findMarkerBloggerMessagesRange((bloggerLog.getCurrentBloggerId()-1)*taskParam.getRecords(),taskParam.getRecords());

    }

    @Override
    public void sendMsg(List<BloggerMessage> messages, String topic) {
        messages.forEach(msg->{
            msg.setRetryTimes(appConfig.getRetryBloggerTimes());
            msg.setRetryLimitTimes(appConfig.getRetryBloggerTimes());
            kafkaTemplateBlogger.send(topic,msg);
        });
    }
}
